﻿using System;
using System.Collections.Generic;
using System.Threading;
using System.Threading.Tasks;
using System.ServiceModel;
using System.Linq;
using XchangeStreamer.OutputSdk.MarketDataFeedService;
using VFNStockTrader.BSL.MarketDataFeed.DataContracts;
using VFNStockTrader.BSL.Utilities.CompressionManager;
using XchangeStreamer.OutputSdk.Internal;

namespace XchangeStreamer.OutputSdk
{
    public sealed class Connector : IDisposable
    {
        #region Private Fields

        #region Basic

        private static Connector connection;
        private MarketDataFeedClient client;
        private Callback callback;
        private Timer timer;
        private long disposed = 0;
        private long connected = 0;
        private long error = 0;
        private NetTcpBinding netTcp;
        private EndpointAddress address;
        private object locker = new object();
        
        #endregion

        #endregion

        #region Private Const. / Dest.

        private Connector()
        {
        }

        ~Connector()
        {
            this.Disconnect(false);
        }

        #endregion

        #region Public Properties

        public static Connector Connection
        {
            get
            {
                if (connection == null)
                {
                    lock (typeof(Connector))
                    {
                        if (connection == null)
                        {
                            connection = new Connector();
                        }
                    }
                }
                return connection;
            }
        }

        #endregion

        #region Public Methods

        public void Connect()
        {
            if (Interlocked.CompareExchange(ref this.connected, 1, 0) == 0)
            {
                if (this.marketManager == null)
                {
                    this.marketManager = new MarketManager();
                }

                Task.Factory.StartNew(() =>
                {
                    lock (this.locker)
                    {
                        this.Init();

                        this.client = new MarketDataFeedClient(new InstanceContext(this.callback), this.netTcp, this.address);
                        this.Start();

                        Interlocked.Exchange(ref this.disposed, 0);
                    }
                    this.Consume();
                });
            }
        }

        public void Disconnect()
        {
            this.Disconnect(true);
        }

        #endregion

        #region Private Methods

        private void Init()
        {
            this.callback = new Callback();

            // NetTcp Binding
            this.netTcp = new NetTcpBinding(SecurityMode.None, true);

            this.netTcp.OpenTimeout = TimeSpan.FromDays(1);
            this.netTcp.CloseTimeout = TimeSpan.FromDays(1);
            this.netTcp.SendTimeout = TimeSpan.FromDays(1);
            this.netTcp.ReceiveTimeout = TimeSpan.FromDays(5);

            this.netTcp.Security.Mode = SecurityMode.None;
            this.netTcp.Security.Message.ClientCredentialType = MessageCredentialType.Windows;
            this.netTcp.Security.Transport.ClientCredentialType = TcpClientCredentialType.Windows;

            this.netTcp.ListenBacklog = int.MaxValue;
            this.netTcp.MaxBufferPoolSize = int.MaxValue;
            this.netTcp.MaxBufferSize = int.MaxValue;

            /*
             * (Not Supported on Mono)
             * 
            this.netTcp.MaxConnections = 1000;
            */

            this.netTcp.MaxReceivedMessageSize = int.MaxValue;
            this.netTcp.PortSharingEnabled = false;
            this.netTcp.TransferMode = TransferMode.Buffered;

            this.netTcp.ReaderQuotas.MaxArrayLength = int.MaxValue;
            this.netTcp.ReaderQuotas.MaxBytesPerRead = int.MaxValue;
            this.netTcp.ReaderQuotas.MaxDepth = int.MaxValue;
            this.netTcp.ReaderQuotas.MaxNameTableCharCount = int.MaxValue;
            this.netTcp.ReaderQuotas.MaxStringContentLength = int.MaxValue;

            /*
             * (Not Supported on Mono)
             * 
            this.netTcp.ReliableSession.Enabled = true;
            this.netTcp.ReliableSession.InactivityTimeout = TimeSpan.FromDays(5);
            this.netTcp.ReliableSession.Ordered = true;
            */

            this.netTcp.ReliableSession.Enabled = false;
            this.netTcp.ReliableSession.InactivityTimeout = TimeSpan.FromDays(5);
            this.netTcp.ReliableSession.Ordered = true;

            this.address = new EndpointAddress("net.tcp://217.52.114.5:7002/MarketDataFeedService");
        }

        private void Start()
        {
            bool opened = false;
            while (!opened)
            {
                try
                {
                    this.client.Open();
                    opened = true;
                }
                catch { }
            }

            this.client.InnerDuplexChannel.Faulted += new EventHandler(this.InnerDuplexChannel_Faulted);

            Console.WriteLine("Connected.");
            this.client.Subscribe();
            Console.WriteLine("Subscribed.");
            this.ProcessServerPendingMessages();

            this.timer = new Timer(new TimerCallback(this.TimerCallback), null, 2 * 60 * 1000, 5 * 60 * 1000);
        }

        private void Disconnect(bool suppressFinalize)
        {
            if (Interlocked.CompareExchange(ref this.disposed, 1, 0) == 0)
            {
                lock (this.locker)
                {
                    if (this.timer != null)
                    {
                        this.timer.Dispose();
                        this.timer = null;
                    }

                    try
                    {
                        this.client.Unsubscribe();
                    }
                    catch { }
                    finally
                    {
                        if (this.client != null)
                        {
                            this.client.Close();
                            this.client = null;
                        }

                        if (this.callback != null)
                        {
                            this.callback.Dispose();
                            this.callback = null;
                        }

                        if (this.marketManager != null)
                        {
                            //////////////////////////////////////////////////////////////////////////////////////////////
                            /////////////////////////////////Must save this object here;//////////////////////////////////
                            /////////////////////////////////////////////////////////////////////////////////////////////
                            this.marketManager = null;
                        }
                        
                        Interlocked.Exchange(ref this.connected, 0);

                        if (suppressFinalize)
                        {
                            GC.SuppressFinalize(this);
                        }
                    }
                }
            }
        }

        private void Consume()
        {
            foreach (var message in this.callback.GetEnumerator())
            {
                switch (message.Header.Identity.Split(':')[0])
                {
                    case "Quote":
                        this.ProcessOrders(message);
                        break;
                    case "Tick":
                        this.ProcessTrades(message);
                        break;
                    case "SymbolSummary":
                        this.ProcessSymbolSummary(message);
                        break;
                    case "SymbolInfo":
                        this.ProcessSymbolInfo(message);
                        break;
                    case "Investor":
                        this.ProcessInvestors(message);
                        break;
                    case "Institations":
                        this.ProcessRetialRatios(message);
                        break;
                    case "News":
                        this.ProcessNews(message);
                        break;
                    case "Indices":
                        this.ProcessIndex(message);
                        break;
                    case "Sectors":
                        this.ProcessSectors(message);
                        break;
                    case "Bulletin":
                        this.ProcessBulletins(message);
                        break;
                    case "BasicInfo":
                        this.ProcessBasicInfo(message);
                        break;
                    case "CaseSectorIndex":
                        this.ProcessSectorIndex(message);
                        break;
                    case "Symbol52HighLow":
                        this.ProcessSymbol52HighLow(message);
                        break;
                    default:
                        break;
                }
            }
        }

        private void ProcessServerPendingMessages()
        {
            var tradesTask = Task.Factory.StartNew(this.TradesTask);
            var ordersTask = Task.Factory.StartNew(this.OrdersTask);
            var indexTask = Task.Factory.StartNew(this.IndexTask);
            var symbolInfoTask = Task.Factory.StartNew(this.SymbolInfoTask);
            var symbolSummaryTask = Task.Factory.StartNew(this.SymbolSummaryTask);
            var basicInfoTask = Task.Factory.StartNew(this.BasicInfoTask);
            var sectorsTask = Task.Factory.StartNew(this.SectorsTask);
            var symbol52HighLowTask = Task.Factory.StartNew(this.Symbol52HighLowTask);
            var bulletinsTask = Task.Factory.StartNew(this.BulletinsTask);
            var newsTask = Task.Factory.StartNew(this.NewsTask);
            var caseSectorIndexTask = Task.Factory.StartNew(this.CaseSectorIndexTask);
            var investorsTask = Task.Factory.StartNew(this.InvestorsTask);
            var retailRatiosTask = Task.Factory.StartNew(this.RetailRatiosTask);

            Task.WaitAll(tradesTask, ordersTask, indexTask, symbolInfoTask, symbolSummaryTask, basicInfoTask, sectorsTask,
                            symbol52HighLowTask, bulletinsTask, newsTask, caseSectorIndexTask, investorsTask, retailRatiosTask);

            Console.WriteLine("Done");
        }

        private void TimerCallback(object obj)
        {
            if (Interlocked.Read(ref this.error) == 0)
            {
                try
                {
                    this.client.IsOnline();
                }
                catch { }
            }
        }

        private void InnerDuplexChannel_Faulted(object sender, EventArgs e)
        {
            if (Interlocked.CompareExchange(ref this.error, 1, 0) == 0)
            {
                this.client.InnerDuplexChannel.Faulted -= new EventHandler(this.InnerDuplexChannel_Faulted);
                if (this.disposed == 1 || this.connected == 1)
                {
                    if (this.timer != null)
                    {
                        this.timer.Dispose();
                        this.timer = null;
                    }

                    this.Start();
                    Interlocked.Exchange(ref this.error, 0);
                }
            }
        }

        #endregion

        #region Download Tasks

        private void TradesTask()
        {
            //return;
            var message = this.client.GetTradesIntraDayHistory(string.Empty, DateTime.MinValue);
            this.ProcessTrades(message);
        }

        private void OrdersTask()
        {
            //return;
            var message = this.client.GetOrdersIntraDayHistory(int.MinValue, DateTime.MinValue);
            this.ProcessOrders(message);
        }

        private void IndexTask()
        {
            //return;
            var message = this.client.GetCaseIndexIntraDayHistory(DateTime.MinValue);
            this.ProcessIndex(message);
        }

        private void SymbolInfoTask()
        {
            //return;
            var message = this.client.GetSymbolInfoIntraDayHistory(DateTime.MinValue);
            this.ProcessSymbolInfo(message);
        }

        private void SymbolSummaryTask()
        {
            //return;
            var message = this.client.GetSymbolSummaryIntraDayHistory(DateTime.MinValue);
            this.ProcessSymbolSummary(message);
        }

        private void BasicInfoTask()
        {
            //return;
            var message = this.client.GetBasicInfoIntraDayHistory();
            this.ProcessBasicInfo(message);
        }

        private void SectorsTask()
        {
            //return;
            var message = this.client.GetSectorIntraDayHistory(DateTime.MinValue);
            this.ProcessSectors(message);
        }

        private void Symbol52HighLowTask()
        {
            //return;
            var message = this.client.GetSymbol52HighLowIntraDayHistory(DateTime.MinValue);
            this.ProcessSymbol52HighLow(message);
        }

        private void BulletinsTask()
        {
            //return;
            var message = this.client.GetBulletinIntraDayHistory(decimal.MinValue, DateTime.MinValue);
            this.ProcessBulletins(message);
        }

        private void NewsTask()
        {
            //return;
            var message = this.client.GetNewsIntraDayHistory(decimal.MinValue, DateTime.MinValue);
            this.ProcessNews(message);
        }

        private void CaseSectorIndexTask()
        {
            //return;
            var message = this.client.GetCaseSectorIndexIntraDayHistory(DateTime.MinValue);
            this.ProcessSectorIndex(message);
        }

        private void InvestorsTask()
        {
            //return;
            var message = this.client.GetInvestorIntraDayHistory(DateTime.MinValue);
            this.ProcessInvestors(message);
        }

        private void RetailRatiosTask()
        {
            //return;
            var message = this.client.GetReatailRatiosIntraDayHistory(TimeSpan.MinValue);
            this.ProcessRetialRatios(message);
        }

        #endregion

        private DateTime lastTrade;
        private DateTime lastBid;
        private DateTime lastAsk;

        private List<int> processedOrders = new List<int>();
        private int maxRec = 0;
        private int minRec = 0;

        #region Processing Tasks

        private void ProcessTrades(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawTrade>.Decompress<List<RawTrade>>(message.Body);

            if (data == null)
            {
                return;
            }

            if (data.Count > 0)
            {
                //DateTime maxDate = data.Max(t => t.Date.Date);
                //if (this.lastTrade.Date != DateTime.MinValue.Date)
                //{
                //    if (this.lastTrade.Date < maxDate.Date)
                //    {
                //        this.trades.Clear();
                //        this.lastTrade = maxDate;
                //    }
                //}
                //else
                //{
                //    this.lastTrade = maxDate;
                //}
            }
            else
            {
                return;
            }

            var groups = from trade in data
                         group trade by trade.SymbolCode into temp
                         select temp;

            foreach (var group in groups)
            {
                this.marketManager.Current.Find(group.Key).Process(group);
            }

            data = null;
            //Console.WriteLine("{0} Count: {1}", "Trades", data.Count);
        }

        private void ProcessOrders(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawOrder>.Decompress<List<RawOrder>>(message.Body);

            if (data == null)
            {
                return;
            }

            if (data.Count > 0)
            {
                //var maxDate = data.Max(o => o.Date.Date);

                //if (this.lastBid.Date != DateTime.MinValue.Date)
                //{
                //    if (this.lastBid.Date < maxDate.Date)
                //    {
                //        this.bids.Clear();
                //        data.RemoveAll(o => o.OrderType == QuoteType.Buy && o.OrderAction != QuoteStatus.Open);
                //        this.lastBid = maxDate;
                //    }
                //}
                //else
                //{
                //    data.RemoveAll(o => o.OrderType == QuoteType.Buy && o.OrderAction != QuoteStatus.Open);
                //    this.lastBid = maxDate;
                //}

                //if (this.lastAsk.Date != DateTime.MinValue.Date)
                //{
                //    if (this.lastAsk.Date < maxDate.Date)
                //    {
                //        this.asks.Clear();
                //        data.RemoveAll(o => o.OrderType == QuoteType.Sell && o.OrderAction != QuoteStatus.Open);
                //        this.lastAsk = maxDate;
                //    }
                //}
                //else
                //{
                //    data.RemoveAll(o => o.OrderType == QuoteType.Sell && o.OrderAction != QuoteStatus.Open);
                //    this.lastAsk = maxDate;
                //}

                int max = data.Max(o => o.RecSerial);
                int min = data.Min(o => o.RecSerial);

                if (this.maxRec >= min)
                {
                    if (min >= this.minRec && max <= this.maxRec)
                    {
                        return;
                    }
                    else
                    {
                        if (this.processedOrders.Contains(min) && this.processedOrders.Contains(max))
                        {
                            return;
                        }
                        else
                        {
                            throw new InvalidOperationException("Messages out of order.");
                        }
                    }
                }
                this.maxRec = max;
                this.minRec = min;
            }
            else
            {
                return;
            }

            this.processedOrders.AddRange(from order in data
                                          select order.RecSerial);

            #region Bids

            var bidsTask = Task.Factory.StartNew(() =>
            {
                var newBids = data.FindAll(o => o.OrderType == QuoteType.Buy);

                var bidsGroup = from bid in newBids
                                group bid by bid.SymbolCode into temp
                                select temp;

                foreach (var group in bidsGroup)
                {
                    this.marketManager.Current.Find(group.Key).Process(group, QuoteType.Buy);
                }
            });

            #endregion

            #region Asks

            var newAsks = data.FindAll(o => o.OrderType == QuoteType.Sell);

            var asksGroup = from ask in newAsks
                            group ask by ask.SymbolCode into temp
                            select temp;

            foreach (var group in asksGroup)
            {
                this.marketManager.Current.Find(group.Key).Process(group, QuoteType.Sell);
            }

            Task.WaitAll(bidsTask);

            #endregion

            data = null;
            //Console.WriteLine("{0} Count: {1}", "Orders", data.Count);
        }

        private void ProcessIndex(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawMainIndex>.Decompress<List<RawMainIndex>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                //this.indexes.Add(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "Index", data.Count);
        }

        private void ProcessSymbolInfo(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawSymbolInfo>.Decompress<List<RawSymbolInfo>>(message.Body);

            if (data == null)
            {
                return;
            }

            //for (int i = 0; i < data.Count; i++)
            //{
            //    this.symbolsInfo.Add(data[i]);
            //}

            for (int i = 0; i < data.Count; i++)
            {
                XchangeStreamer.OutputSdk.ViewModel.Symbol symbol = new XchangeStreamer.OutputSdk.ViewModel.Symbol();
                symbol.Arb_Name = data[i].ArbName;
                symbol.Isin_Code = data[i].SymbolCode;
                symbol.Eng_Name = data[i].EngName;

                this.marketManager.Add(symbol);
            }

            data = null;
            //Console.WriteLine("{0} Count: {1}", "SymbolInfo", data.Count);
        }

        private void ProcessSymbolSummary(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawSymbolSummary>.Decompress<List<RawSymbolSummary>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                this.marketManager.Current.Find(data[i].SymbolCode).Process(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "SymbolSummary", data.Count);
        }

        private void ProcessBasicInfo(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<MarketBasicInfo>.Decompress<MarketBasicInfo>(message.Body);

            if (data == null)
            {
                return;
            }

            //this.basicInfo.Add(data);
            //Console.WriteLine("{0} Time to Open: {1}", "BasicInfo", data.TimeToOpen);
        }

        private void ProcessSectors(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawSector>.Decompress<List<RawSector>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                //this.sectors.Add(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "Sectors", data.Count);
        }

        private void ProcessSymbol52HighLow(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawSymbol52HighLow>.Decompress<List<RawSymbol52HighLow>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                this.marketManager.Current.Find(data[i].SymbolCode).Process(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "Symbol52HighLow", data.Count);
        }

        private void ProcessBulletins(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawBulletin>.Decompress<List<RawBulletin>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                this.marketManager.Current.Find(data[i].SymbolCode).Process(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "Bulletins", data.Count);
        }

        private void ProcessNews(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawNews>.Decompress<List<RawNews>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                this.marketManager.Current.Find(data[i].SymbolCode).Process(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "News", data.Count);
        }

        private void ProcessSectorIndex(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawCaseSectorIndex>.Decompress<List<RawCaseSectorIndex>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                //this.sectorsIndex.Add(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "SectorIndex", data.Count);
        }

        private void ProcessInvestors(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawInvestor>.Decompress<List<RawInvestor>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                //this.investors.Add(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "Investors", data.Count);
        }

        private void ProcessRetialRatios(MarketDataFeedService.Message message)
        {
            if (message == null)
            {
                return;
            }
            var data = CompressionManager<RawInstitationsAndReatailRatios>.Decompress<List<RawInstitationsAndReatailRatios>>(message.Body);

            if (data == null)
            {
                return;
            }

            for (int i = 0; i < data.Count; i++)
            {
                //this.retialRatios.Add(data[i]);
            }
            data = null;
            //Console.WriteLine("{0} Count: {1}", "RetialRatios", data.Count);
        }

        #endregion

        #region IDisposable Members

        public void Dispose()
        {
            this.Disconnect(true);
        }

        #endregion

        #region Market Based Interface

        private MarketManager marketManager;
        public IMarket Market
        {
            get
            {
                return this.marketManager.Current;
            }
        }

        #endregion
    }
}